package DianShang_2024.ds_server.clean

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{current_timestamp, date_format, lit, to_timestamp}

object extract03 {
  def main(args: Array[String]): Unit = {
    /*
      3、将ods库中order_master表昨天的分区（任务一生成的分区）数据抽取到dwd库中fact_order_master的动态分区表，分区字段为etl_date，类型为String，
      取create_time值并将格式转换为yyyyMMdd，并添加dwd_insert_user、dwd_insert_time、dwd_modify_user、dwd_modify_time四列，其
      中dwd_insert_user、dwd_modify_user均填写“user1”，dwd_insert_time、dwd_modify_time均填写当前操作时间，并进行数据类型转换，需要过滤掉city字段
      长度大于8。使用hive cli执行show partitions dwd.fact_order_info命令；
    提示：若完成实时任务二中的第3小题后，可将hbase中的数据合并后在进行清洗，清洗后数据添加至DWD中
     */

    //  准备sparksql的环境
    val spark=SparkSession.builder()
      .master("local[*]")
      .appName("数据清洗第三题")
      .enableHiveSupport()
      .getOrCreate()

    spark.sql("use dwd_server")
    //  将hive的分区模式设置为非严格模式(启动动态分区)  dynamic:动态的
    spark.sql("set hive.exec.dynamic.partition.mode=nonstrict")

    /*
           处理逻辑1.首先对ods_server.order_master01表进行添加字段,修改create_time字段格式，过滤city大于8的操作
                  2.对数据进行插入(动态分区)
     */
    //      因为一些原因，具体不清楚，反正不使用string数据类型会报错
    //  去虚拟机hive的shell操作执行: alter table fact_order_master change column dwd_insert_time dwd_insert_time string;
    //                            alter table fact_order_master change column dwd_modify_time dwd_modify_time string;

spark.sql(
  """
    |select
    |order_id ,
    |order_sn ,
    |customer_id ,
    |shipping_user ,
    |province ,
    |city ,
    |address ,
    |order_source ,
    |payment_method ,
    |order_money ,
    |district_money ,
    |shipping_money ,
    |payment_money ,
    |shipping_comp_name ,
    |shipping_sn ,
    |substr(create_time,1,8) as create_time ,
    |shipping_time ,
    |pay_time ,
    |receive_time ,
    |order_status ,
    |order_point ,
    |invoice_title ,
    |modified_time,
    |etl_date
    |from ods_server.order_master01
    |where etl_date='20240311'  and length(city) <= 8
    |""".stripMargin)
  .withColumn("dwd_insert_user",lit("user1"))
  .withColumn(
    "dwd_insert_time",
    to_timestamp(date_format(current_timestamp(),"yyyy-MM-dd HH:mm:ss"),"yyyy-MM-dd HH:mm:ss").cast("timestamp")
  )
  .withColumn("dwd_modify_user",lit("user1"))
  .withColumn(
    "dwd_modify_time",
    to_timestamp(date_format(current_timestamp(),"yyyy-MM-dd HH:mm:ss"),"yyyy-MM-dd HH:mm:ss").cast("timestamp")
  )
  .createOrReplaceTempView("temp_table")

//    spark.sql("desc temp_table").show
//    spark.sql("select * from temp_table").show


    //  根据条件合并数据
    spark.sql(
      """
        |insert into table dwd_server.fact_order_master
        |partition(etl_date)
        |select
        |*
        |from temp_table
        |""".stripMargin)

//    spark.sql("select  create_time from dwd_server.fact_order_master ").show






    //  关闭sparksql的环境
     spark.close()
  }

}
